package com.steakliu.sun.protocol.future;

import com.steakliu.sun.protocol.*;
import org.apache.commons.lang3.ObjectUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 功能说明：
 * <p>
 * Original @Author: steakliu-刘牌, 2022-12-11  17:34
 * <p>
 * Copyright (C)2020-2022  steakliu All rights reserved.
 */
public class SunFuture extends CompletableFuture<Object> {

    private final Sync sync;

    private SunProtocol<SunResponse> responseRpcProtocol;
    /**
     * 调用开始时间
     */
    private long callStartTime;
    /**
     * 调用超时时间
     */
    private long callTimeOut;

    public SunFuture() {
        this.sync = new Sync();
    }

    @Override
    public boolean isDone() {
        return sync.isDone();
    }

    public static SunFuture getInstance() {
        return new SunFuture();
    }

    /**
     * 获取Future结果，如果获取不到，则阻塞，知道获取为止
     *
     * @return
     */
    @Override
    public Object get() {
        sync.acquire(-1);
        return this.handleResult();
    }

    @Override
    public Object get(long timeout, TimeUnit unit) throws InterruptedException {
        boolean success = sync.tryAcquireNanos(-1, unit.toNanos(timeout));
        if (success) {
            return this.handleResult();
        } else {
            throw new RuntimeException("call rpc service time out");
        }
    }

    private Object handleResult() {
        if (this.responseRpcProtocol != null) {
            SunResponse body = this.responseRpcProtocol.getBody();
            MsgHeader header = this.responseRpcProtocol.getMsgHeader();
            if (ObjectUtils.isNotEmpty(body.getErrorMsg()) && header.getStatus() == MsgStatus.FAIL.getStatus()) {
                throw new RuntimeException(body.getErrorMsg());
            } else {
                return body.getResult();
            }
        } else {
            return null;
        }
    }

    @Override
    public boolean isCancelled() {
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        throw new UnsupportedOperationException();
    }

    public void done(SunProtocol<SunResponse> responseRpcProtocol) {
        this.responseRpcProtocol = responseRpcProtocol;
        sync.release(1);
    }


    static class Sync extends AbstractQueuedSynchronizer {

        @Override
        protected boolean tryAcquire(int acquires) {
            return getState() == 1;
        }

        @Override
        protected boolean tryRelease(int releases) {
            if (getState() == 0) {
                return compareAndSetState(0, 1);
            }
            return false;
        }

        public boolean isDone() {
            getState();
            return getState() == 1;
        }
    }
}
